package Transforming_Observables

import (
	"fmt"
	"github.com/reactivex/rxgo/v2"
	"time"
)

// 定期聚集一些观察者集合中的对象打包成一个组合，然后发送这些组合，而不是一个个的发送对象
func Demo_BUffer() {
	Demo_BufferWithCount()
	Demo_BufferWithTime()
	Demo_BufferWithTimeOrCount()
}

func Demo_BufferWithCount() {
	fmt.Println("Demo_BufferWithCount###########################")
	// 按照数量打包 BufferWithCount
	observable := rxgo.Just(1, 2, 3, 4)().BufferWithCount(3)
	for item := range observable.Observe() {
		if item.Error() {
			fmt.Println("订阅：", item.E)
		}
		fmt.Println("订阅：", item.V)
	}
}

func Demo_BufferWithTime() {
	fmt.Println("Demo_BufferWithTime###########################")
	// 按照时间打包 BufferWithTime
	// Create the producer
	ch := make(chan rxgo.Item, 1)
	go func() {
		i := 0
		for range time.Tick(time.Second) {
			ch <- rxgo.Of(i)
			i++
		}
	}()
	observable := rxgo.FromChannel(ch).
		BufferWithTime(rxgo.WithDuration(3*time.Second), nil)

	for item := range observable.Observe() {
		if item.Error() {
			fmt.Println("订阅：", item.E)
		}
		fmt.Println("订阅：", item.V)
	}
}

func Demo_BufferWithTimeOrCount() {
	fmt.Println("BufferWithTimeOrCount###########################")
	// 按照时间打包 BufferWithTime
	// Create the producer
	// Create the producer
	ch := make(chan rxgo.Item, 1)
	go func() {
		i := 0
		for range time.Tick(time.Second) {
			ch <- rxgo.Of(i)
			i++
		}
	}()

	observable := rxgo.FromChannel(ch).
		BufferWithTimeOrCount(rxgo.WithDuration(3*time.Second), 2)

	for item := range observable.Observe() {
		if item.Error() {
			fmt.Println("订阅：", item.E)
		}
		fmt.Println("订阅：", item.V)
	}
}
